package com.lin.service.impl;

import com.lin.entity.Msg;
import com.lin.thread.AsyncManager;
import com.lin.thread.task.InHbaseByMutatorTask;
import com.lin.thread.task.InHbaseByPutTask;
import com.lin.tool.ExcelReader;
import com.lin.tool.HBaseConnection;
import com.lin.tool.MoMoMsgGen;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.omg.CORBA.PRIVATE_MEMBER;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.CountDownLatch;

/**
 * TODO:多线程写入
 * TODO:hbase集群均衡
 */
@Service
public class InsertIntoHbase {

    private static Set<String> set = new HashSet<>();


//    public static void main(String[] args) {
//        try {
////            mutatorInsertByThread();
//        } catch (ParseException | IOException e) {
//            e.printStackTrace();
//        }
//    }

    /**
     * 测试插入10w数据速度 put方式-单线程
     * 48470 耗时：91165
     * @throws ParseException
     */
    public void putInsert() throws ParseException, IOException {
        Map<String, List<String>> resultMap = ExcelReader.readXlsx("E:\\ljl\\DataToHbase\\data\\测试数据集.xlsx", "陌陌数据");

        // hbase连接
        Connection connection = HBaseConnection.connection;
        // table连接
        Table table = connection.getTable(TableName.valueOf("MOMO_CHAT:MSG"));
        int i = 1;
        int max = 50001;

        long start = System.currentTimeMillis();
        while (i <= max){
            // 要插入的一条数据
            Msg oneMsg = MoMoMsgGen.getOneMsg(resultMap);
            byte[] rowKey = MoMoMsgGen.getRowkey(oneMsg,set);

            // 构建一个put
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_time"),Bytes.toBytes(oneMsg.getMsg_time()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_nickyname"),Bytes.toBytes(oneMsg.getSender_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_account"),Bytes.toBytes(oneMsg.getSender_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_sex"),Bytes.toBytes(oneMsg.getSender_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_ip"),Bytes.toBytes(oneMsg.getSender_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_os"),Bytes.toBytes(oneMsg.getSender_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_phone_type"),Bytes.toBytes(oneMsg.getSender_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_network"),Bytes.toBytes(oneMsg.getSender_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_gps"),Bytes.toBytes(oneMsg.getSender_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_nickyname"),Bytes.toBytes(oneMsg.getReceiver_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_ip"),Bytes.toBytes(oneMsg.getReceiver_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_account"),Bytes.toBytes(oneMsg.getReceiver_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_os"),Bytes.toBytes(oneMsg.getReceiver_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_phone_type"),Bytes.toBytes(oneMsg.getReceiver_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_network"),Bytes.toBytes(oneMsg.getReceiver_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_gps"),Bytes.toBytes(oneMsg.getReceiver_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_sex"),Bytes.toBytes(oneMsg.getReceiver_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_type"),Bytes.toBytes(oneMsg.getMsg_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("distance"),Bytes.toBytes(oneMsg.getDistance()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("message"),Bytes.toBytes(oneMsg.getMessage()));

            table.put(put);
            i++;
        }

        long end = System.currentTimeMillis();
        System.out.println("put方式-单线程 耗时：" + (end - start));
        table.close();
    }

    /**
     * 测试插入10w数据速度 puts方式-单线程
     * 18266 耗时：16143
     * @throws ParseException
     */
    public void putsInsert() throws ParseException, IOException {
        Map<String, List<String>> resultMap = ExcelReader.readXlsx("E:\\ljl\\DataToHbase\\data\\测试数据集.xlsx", "陌陌数据");

        // hbase连接
        Connection connection = HBaseConnection.connection;
        // table连接
        Table table = connection.getTable(TableName.valueOf("MOMO_CHAT:MSG"));
        int i = 1;
        int max = 50001;
        // 批量put
        List<Put> puts = new ArrayList<>();

        long start = System.currentTimeMillis();
        while (i <= max){
            // 要插入的一条数据
            Msg oneMsg = MoMoMsgGen.getOneMsg(resultMap);
            byte[] rowKey = MoMoMsgGen.getRowkey(oneMsg,set);

            // 构建一个put
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_time"),Bytes.toBytes(oneMsg.getMsg_time()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_nickyname"),Bytes.toBytes(oneMsg.getSender_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_account"),Bytes.toBytes(oneMsg.getSender_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_sex"),Bytes.toBytes(oneMsg.getSender_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_ip"),Bytes.toBytes(oneMsg.getSender_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_os"),Bytes.toBytes(oneMsg.getSender_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_phone_type"),Bytes.toBytes(oneMsg.getSender_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_network"),Bytes.toBytes(oneMsg.getSender_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_gps"),Bytes.toBytes(oneMsg.getSender_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_nickyname"),Bytes.toBytes(oneMsg.getReceiver_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_ip"),Bytes.toBytes(oneMsg.getReceiver_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_account"),Bytes.toBytes(oneMsg.getReceiver_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_os"),Bytes.toBytes(oneMsg.getReceiver_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_phone_type"),Bytes.toBytes(oneMsg.getReceiver_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_network"),Bytes.toBytes(oneMsg.getReceiver_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_gps"),Bytes.toBytes(oneMsg.getReceiver_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_sex"),Bytes.toBytes(oneMsg.getReceiver_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_type"),Bytes.toBytes(oneMsg.getMsg_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("distance"),Bytes.toBytes(oneMsg.getDistance()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("message"),Bytes.toBytes(oneMsg.getMessage()));

            puts.add(put);
            if(i % 100000 == 0){
                table.put(puts);
                puts.clear();
            }
            i++;
        }

        if(puts.size() > 0) table.put(puts);

        long end = System.currentTimeMillis();
        System.out.println("puts方式-单线程 耗时：" + (end - start));
        table.close();
    }


    /**
     * 测试插入10w数据速度 BufferedMutator方式-单线程
     *
     * BufferedMutator：客户端提供了缓冲区，put的数据放在缓存区中，当达到一定量后或用户提交才一次性提交到hbase服务器。
     * mutate方法提交数据，flush方法可以强制刷新缓冲区提交数据
     *
     * 26659 耗时：3173
     *
     * @throws ParseException
     */
    public void mutatorInsert() throws ParseException, IOException {
        Map<String, List<String>> resultMap = ExcelReader.readXlsx("E:\\ljl\\DataToHbase\\data\\测试数据集.xlsx", "陌陌数据");

        // hbase连接
        Connection connection = HBaseConnection.connection;

        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf("MOMO_CHAT:MSG"));
        params.writeBufferSize(6 * 1024 * 1024);

        BufferedMutator mutator = connection.getBufferedMutator(params);


        int i = 1;
        int max = 50001;
        // 批量put
        List<Put> puts = new ArrayList<>();

        long start = System.currentTimeMillis();
        while (i <= max){
            // 要插入的一条数据
            Msg oneMsg = MoMoMsgGen.getOneMsg(resultMap);
            byte[] rowKey = MoMoMsgGen.getRowkey(oneMsg,set);

            // 构建一个put
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_time"),Bytes.toBytes(oneMsg.getMsg_time()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_nickyname"),Bytes.toBytes(oneMsg.getSender_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_account"),Bytes.toBytes(oneMsg.getSender_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_sex"),Bytes.toBytes(oneMsg.getSender_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_ip"),Bytes.toBytes(oneMsg.getSender_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_os"),Bytes.toBytes(oneMsg.getSender_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_phone_type"),Bytes.toBytes(oneMsg.getSender_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_network"),Bytes.toBytes(oneMsg.getSender_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_gps"),Bytes.toBytes(oneMsg.getSender_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_nickyname"),Bytes.toBytes(oneMsg.getReceiver_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_ip"),Bytes.toBytes(oneMsg.getReceiver_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_account"),Bytes.toBytes(oneMsg.getReceiver_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_os"),Bytes.toBytes(oneMsg.getReceiver_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_phone_type"),Bytes.toBytes(oneMsg.getReceiver_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_network"),Bytes.toBytes(oneMsg.getReceiver_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_gps"),Bytes.toBytes(oneMsg.getReceiver_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_sex"),Bytes.toBytes(oneMsg.getReceiver_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_type"),Bytes.toBytes(oneMsg.getMsg_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("distance"),Bytes.toBytes(oneMsg.getDistance()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("message"),Bytes.toBytes(oneMsg.getMessage()));

            puts.add(put);
            if(i % 5000 == 0){
                long s = System.currentTimeMillis();
                mutator.mutate(puts);
                mutator.flush();
                long e = System.currentTimeMillis();
                System.out.println("1w耗时：" + (e - s));
                puts.clear();
            }
            i++;
        }

        if(puts.size() > 0){
            mutator.mutate(puts);
            mutator.flush();
        }

        long end = System.currentTimeMillis();
        System.out.println(puts.size() + "耗时：" + (end - start));
        System.out.println(set.size());
        set.clear();
        mutator.close();
    }

    /**
     * 测试插入10w数据速度 BufferedMutator方式-多线程
     * 100000(3个region) 耗时：3098
     * 100000(6个region) 耗时：1849
     * @throws ParseException
     */
    public void mutatorInsertByThread() throws ParseException, IOException {
        Map<String, List<String>> resultMap = ExcelReader.readXlsx("E:\\ljl\\DataToHbase\\data\\测试数据集.xlsx", "陌陌数据");

        // hbase连接
        Connection connection = HBaseConnection.connection;


        int i = 1;
        int max = 100000;
        int batchSize = 10000;

        int count = 0;
        if(max % batchSize == 0) count = max / batchSize;
        else count = (max / batchSize) + 1;
        System.out.println(count);

        CountDownLatch countDownLatch = new CountDownLatch(count);
        // 批量put
        List<Put> puts = new ArrayList<>();

        long start = System.currentTimeMillis();
        while (i <= max){
            // 要插入的一条数据
            Msg oneMsg = MoMoMsgGen.getOneMsg(resultMap);
            byte[] rowKey = MoMoMsgGen.getRowkey(oneMsg,set);

            // 构建一个put
            Put put = new Put(rowKey);

            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mt"),Bytes.toBytes(oneMsg.getMsg_time()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sn"),Bytes.toBytes(oneMsg.getSender_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sa"),Bytes.toBytes(oneMsg.getSender_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("ss"),Bytes.toBytes(oneMsg.getSender_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("si"),Bytes.toBytes(oneMsg.getSender_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("so"),Bytes.toBytes(oneMsg.getSender_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("spt"),Bytes.toBytes(oneMsg.getSender_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sne"),Bytes.toBytes(oneMsg.getSender_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sg"),Bytes.toBytes(oneMsg.getSender_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("rn"),Bytes.toBytes(oneMsg.getReceiver_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("ri"),Bytes.toBytes(oneMsg.getReceiver_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("ra"),Bytes.toBytes(oneMsg.getReceiver_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("ro"),Bytes.toBytes(oneMsg.getReceiver_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("rpt"),Bytes.toBytes(oneMsg.getReceiver_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("rne"),Bytes.toBytes(oneMsg.getReceiver_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("rg"),Bytes.toBytes(oneMsg.getReceiver_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("rs"),Bytes.toBytes(oneMsg.getReceiver_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mtp"),Bytes.toBytes(oneMsg.getMsg_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("dis"),Bytes.toBytes(oneMsg.getDistance()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("message"),Bytes.toBytes(oneMsg.getMessage()));

            puts.add(put);
            if(i % batchSize == 0){
                AsyncManager.me().execute(new InHbaseByMutatorTask(puts,connection,countDownLatch));
                puts = new ArrayList<>();
            }
            i++;
        }

        if(puts.size() > 0) AsyncManager.me().execute(new InHbaseByMutatorTask(puts,connection,countDownLatch));

        // 等待所有任务完成
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        long end = System.currentTimeMillis();
        System.out.println(puts.size() + "耗时：" + (end - start));
        System.out.println(set.size());
        set.clear();
    }

    public void putInsertByThread() throws ParseException, IOException {
        Map<String, List<String>> resultMap = ExcelReader.readXlsx("E:\\ljl\\DataToHbase\\data\\测试数据集.xlsx", "陌陌数据");

        // hbase连接
        Connection connection = HBaseConnection.connection;
        // table连接
//        Table table = connection.getTable(TableName.valueOf("MOMO_CHAT:MSG"));
        int i = 1;
        int max = 50001;
        // 批量put
        List<Put> puts = new ArrayList<>();

        long start = System.currentTimeMillis();
        while (i <= max){
            // 要插入的一条数据
            Msg oneMsg = MoMoMsgGen.getOneMsg(resultMap);
            byte[] rowKey = MoMoMsgGen.getRowkey(oneMsg,set);

            // 构建一个put
            Put put = new Put(rowKey);

            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_time"),Bytes.toBytes(oneMsg.getMsg_time()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_nickyname"),Bytes.toBytes(oneMsg.getSender_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_account"),Bytes.toBytes(oneMsg.getSender_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_sex"),Bytes.toBytes(oneMsg.getSender_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_ip"),Bytes.toBytes(oneMsg.getSender_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_os"),Bytes.toBytes(oneMsg.getSender_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_phone_type"),Bytes.toBytes(oneMsg.getSender_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_network"),Bytes.toBytes(oneMsg.getSender_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sender_gps"),Bytes.toBytes(oneMsg.getSender_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_nickyname"),Bytes.toBytes(oneMsg.getReceiver_nickyname()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_ip"),Bytes.toBytes(oneMsg.getReceiver_ip()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_account"),Bytes.toBytes(oneMsg.getReceiver_account()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_os"),Bytes.toBytes(oneMsg.getReceiver_os()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_phone_type"),Bytes.toBytes(oneMsg.getReceiver_phone_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_network"),Bytes.toBytes(oneMsg.getReceiver_network()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_gps"),Bytes.toBytes(oneMsg.getReceiver_gps()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receiver_sex"),Bytes.toBytes(oneMsg.getReceiver_sex()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("msg_type"),Bytes.toBytes(oneMsg.getMsg_type()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("distance"),Bytes.toBytes(oneMsg.getDistance()));
            put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("message"),Bytes.toBytes(oneMsg.getMessage()));

            puts.add(put);
            if(i % 10000 == 0){
                AsyncManager.me().execute(new InHbaseByPutTask(puts,connection));
                puts.clear();
            }
            i++;
        }

        long end = System.currentTimeMillis();
        System.out.println(puts.size() + "耗时：" + (end - start));
        System.out.println(set.size());
    }
}
